/**
 * Author:yuiop
 * Date:2022/10/20 22:36
 */
package com.devops.admin.config.bean;

import cn.hutool.core.date.DateUtil;
import io.fabric8.kubernetes.api.model.Event;
import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.io.Serializable;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
@Slf4j
@Data
public class K8SClientInform {

    private static Integer macCount = 5;
    private static Long expireTime = 1800L;
    private Map<Integer, InformAndTime> eventInforms = new ConcurrentHashMap<>();
    private Map<Integer, PipedOutputStream> outputStreamMap = new ConcurrentHashMap<>();
    private Map<Integer, PipedInputStream> inputStreamMap = new ConcurrentHashMap<>();

    public void addInform(Integer releaseId, SharedIndexInformer<Event> inform) {
        closeAndRemoveInformByExpireTime();
        InformAndTime watchAndTime = InformAndTime.builder().inform(inform).time(DateUtil.currentSeconds()).build();
        if (eventInforms.get(releaseId) != null) {
            eventInforms.get(releaseId).inform.close();
            eventInforms.remove(releaseId);
            closePiped(releaseId);
        }
        eventInforms.put(releaseId,watchAndTime);
    }

    /**
     * 需要保证线程安全不然多线程下会发生map的快速失败机制
     * @param releaseId
     */
    public void closeAndRemoveInform(Integer releaseId) {
        closeAndRemoveInformByExpireTime();
        if (eventInforms.get(releaseId) != null) {
            eventInforms.get(releaseId).inform.close();
            eventInforms.remove(releaseId);
            closePiped(releaseId);
        }
    }

    public void closeAndRemoveInformByExpireTime() {
        Long currentSeconds = DateUtil.currentSeconds();
        for (Iterator<Map.Entry<Integer, InformAndTime>> it = eventInforms.entrySet().iterator(); it.hasNext();) {
            Map.Entry<Integer, InformAndTime> next = it.next();
            if (currentSeconds - next.getValue().time > expireTime) {
                //watch监听已超过最长时间
                next.getValue().inform.close();
                Integer key = next.getKey();
                closePiped(key);
                it.remove();
            }

        }
    }

    private void closePiped(Integer key) {
        try {
            if (outputStreamMap.get(key) != null){
                outputStreamMap.get(key).close();
                outputStreamMap.remove(key);
            }
            if (inputStreamMap.get(key) != null){
                inputStreamMap.get(key).close();
                inputStreamMap.remove(key);
            }
        }catch (Exception e){
            log.error("关闭管道出错");
            e.printStackTrace();
        }
    }

    public void addPipedMap(Integer releaseId,PipedOutputStream outputStream,PipedInputStream inputStream){
        outputStreamMap.put(releaseId,outputStream);
        inputStreamMap.put(releaseId,inputStream);
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
class InformAndTime implements Serializable {
    SharedIndexInformer<Event> inform;
    Long time;
}
